package com.sunlands.deskmate.message.handler.sender.websocket;

import com.alibaba.fastjson.JSON;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.sunlands.deskmate.config.AsyncExecutorConfig;
import com.sunlands.deskmate.message.entity.Messages;
import com.sunlands.deskmate.channel.ChannelInfo;
import com.sunlands.deskmate.message.handler.sender.RetrySender;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;
import java.nio.channels.ClosedChannelException;
import java.util.concurrent.*;

/**
 * @Author: anjunling@sunlands.com
 * @Description:
 * @Date: Create in 12:48 2018/6/14
 * @Modified By:
 */
@Slf4j
public class RetryWebsocketMessageSender extends AbstractWebsocketMessageSender<Messages>
        implements RetrySender<Messages, Integer> {

    private static volatile Object lock;

    private static final BlockingQueue<RetryWebsocketMessageSender> queue = new LinkedBlockingQueue<>();
    private static ThreadFactory retrySendMessageFactory = new ThreadFactoryBuilder().setNameFormat("retry-send-user-message-pool-%d").build();

    private static final ExecutorService retrySendMessagePool  = new ThreadPoolExecutor(1, 4, 1, TimeUnit.HOURS, new LinkedBlockingDeque<>(1024), retrySendMessageFactory, new AsyncExecutorConfig.AbortPolicy());

    public RetryWebsocketMessageSender(ChannelInfo channelInfo, Messages messages) {
        this.channelInfo = channelInfo;
        this.message = messages;
    }

    @Override
    public void send() {
        super.send();
    }

    @Override
    public void retry() {
        run();
        boolean success = queue.offer(this);
        if (!success) {
            log.debug("RetryWebsocketMessageSender success={}", success);
        }
    }

    @Override
    public Integer strategy() {
        return 2;
    }

    @Override
    protected ChannelFutureListener futureListener() {
        return new RetryWebsocketMessageSender.SendChannelFutureListener(this);
    }

    /**
     * 启动重试器
     */
    private void run() {
        if (lock == null) {
            synchronized (RetryWebsocketMessageSender.class) {
                if (lock == null) {
                    lock = new Object();

                    Worker worker = new Worker();
                    Thread thread = new Thread(worker);
                    thread.start();
                }
            }
        }
    }

    private static class Worker implements Runnable {

        @Override
        public void run() {
            boolean isSuccess = false;
            boolean isFinish = false;
            while (!isSuccess || !isFinish || Thread.currentThread().isInterrupted()) {
                RetryWebsocketMessageSender retryWebsocketMessageSender = null;
                try {
                    retryWebsocketMessageSender = take();
                    if (retryWebsocketMessageSender == null) {
                        log.warn("ws take is empty");
                        continue;
                    }
                    isSuccess = retryWebsocketMessageSender.isSuccess;
                    isFinish = retryWebsocketMessageSender.isFinish;

                    CompletableFuture.completedFuture(retryWebsocketMessageSender).thenAcceptAsync(RetryWebsocketMessageSender::send, retrySendMessagePool);
                } catch (InterruptedException ie) {
                    log.error("retry websocket interrupted exception");
                    Thread.currentThread().interrupt();
                } catch (Exception e) {
                    if (retryWebsocketMessageSender != null) {
                        log.error("ws send fail channelInfo {} messages {}", retryWebsocketMessageSender.channelInfo, JSON.toJSONString(retryWebsocketMessageSender.message));
                    } else {
                        log.error("ws send fail", e);
                    }
                }
            }
        }

        private RetryWebsocketMessageSender take() throws InterruptedException {
            RetryWebsocketMessageSender sender;
            try {
                sender = queue.take();
            } catch (InterruptedException ie) {
                log.warn("take retry");
                throw ie;
            }
            return sender;
        }
    }

    private static class SendChannelFutureListener implements ChannelFutureListener {

        private RetryWebsocketMessageSender retryWebsocketMessageSender;

        public SendChannelFutureListener() {
        }

        public SendChannelFutureListener(RetryWebsocketMessageSender retryWebsocketMessageSender) {
            this.retryWebsocketMessageSender = retryWebsocketMessageSender;
        }

        @Override
        public void operationComplete(ChannelFuture channelFuture) throws Exception {
            synchronized (retryWebsocketMessageSender) {
                try {
                    if (!channelFuture.isSuccess()) {
                        Throwable throwable = channelFuture.cause();
                        if (throwable instanceof ClosedChannelException) {
                            retryWebsocketMessageSender.releaseChannel(channelFuture.channel());

                            log.warn("send channel future close channelInfo {}", retryWebsocketMessageSender.channelInfo);
                            retryWebsocketMessageSender.isFinish = true;
                            retryWebsocketMessageSender.isDone = true;
                        } else {
                            retryWebsocketMessageSender.release();

                            log.error("send channel future exception channelInfo " + retryWebsocketMessageSender.channelInfo, throwable);
                            retryWebsocketMessageSender.isDone = true;
                        }
                    } else {
                        retryWebsocketMessageSender.isSuccess = true;
                        retryWebsocketMessageSender.isFinish = true;
                        //TODO 这个isDone离的也很远
                        retryWebsocketMessageSender.isDone = true;
                    }
                } catch (Exception e) {
                    log.error("send future listener channelInfo " + retryWebsocketMessageSender.channelInfo, e);
                } finally {
                    if (log.isDebugEnabled()) {
                        log.debug("ws sender notify all");
                    }
                    retryWebsocketMessageSender.notifyAll();
                }
            }
        }
    }

}
